import threading
import time

import grpc
import rx, rx.operators as ops
from rx import scheduler

from demo import server_stream_pb2_grpc, server_stream_pb2

# 创建一个 4 线程的调度器
pool_sched = scheduler.ThreadPoolScheduler(max_workers=4)

def to_observable(stream): #把 gRPC 服务端流 -> RxPy Observable
    def subscribe(observer, scheduler=None):
        try:
            for tick in stream:          # 逐条读取
                observer.on_next(tick)
            observer.on_completed()
        except Exception as e:
            observer.on_error(e)
    return rx.create(subscribe)

def handle_stream(stream):
    to_observable(stream).pipe(
        ops.observe_on(pool_sched),
        ops.map(lambda t: t.title),              # 取时间戳
        #ops.buffer_with_time(1.0),            # 每 2 秒聚一批
        ops.filter(lambda batch: batch)       # 去掉空批

    ).subscribe(
        on_next=lambda b: print(threading.current_thread().name,'收到:', b),
        on_error=lambda e: print('err:', e),
        on_completed=lambda: print('stream done')
    )

def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = server_stream_pb2_grpc.SearchStub(channel)
    q1 = server_stream_pb2.Query(keyword="java")
    stream1=stub.SearchStream(q1)
    print('Client start streaming...1')
    q2 = server_stream_pb2.Query(keyword="python")
    stream2=stub.SearchStream(q2)
    print('Client start streaming...2')

    handle_stream(stream1)
    handle_stream(stream2)
    while(True):
        time.sleep(1)


if __name__ == '__main__':
    run()